package com.fpx.metadata;

import com.google.common.collect.Maps;
import lombok.extern.log4j.Log4j2;
import org.springframework.jdbc.core.JdbcTemplate;

import java.sql.ResultSetMetaData;
import java.util.Map;
import java.util.concurrent.RecursiveTask;

@Log4j2
public class QueryTask extends RecursiveTask<Map<Object, Map<String, Object>>> {

    private static final int THRESHOLD = 10000;

    private JdbcTemplate jdbcTemplate;

    private String primaryKey;

    private int start;

    private int end;

    private String sql;

    public QueryTask(JdbcTemplate jdbcTemplate, String primaryKey, int start, int end, String sql) {
        this.jdbcTemplate = jdbcTemplate;
        this.primaryKey = primaryKey;
        this.start = start;
        this.end = end;
        this.sql = sql;
    }

    @Override
    protected Map<Object, Map<String, Object>> compute() {

        Map<Object, Map<String, Object>> result = Maps.newHashMap();
        boolean canComputer = (end - start) <= THRESHOLD;
        if (canComputer) {
            log.info("不需要拆分查询开始了 线程 ：{}", Thread.currentThread().getName());
            String format = String.format(" limit %s,%s", start, end);
            String limitSql = sql + format;
            log.info("{} 线程 执行sql : {}",limitSql);
            return jdbcTemplate.query(limitSql, rs -> {
                ResultSetMetaData metaData = rs.getMetaData();
                Map<Object, Map<String, Object>> rsKeyMap = Maps.newHashMap();
                while (rs.next()) {
                    Map<String, Object> rsMap = Maps.newHashMap();
                    for (int i = 1; i <= metaData.getColumnCount(); i++) {
                        String columnName = metaData.getColumnName(i);
                        rsMap.put(columnName, rs.getObject(i));
                    }
                    rsKeyMap.put(rsMap.get(primaryKey), rsMap);
                }
                return rsKeyMap;
            });
        } else {
            log.info("拆分查询开始了 线程 ：{}", Thread.currentThread().getName());
            //如果任务大于阈值就裂变为两个子任务
            int middle = (start + end) / 2;
            QueryTask leftTask = new QueryTask(jdbcTemplate, primaryKey, start, middle, sql);
            QueryTask rightTask = new QueryTask(jdbcTemplate, primaryKey,middle + 1, end, sql);
            leftTask.fork();
            rightTask.fork();
            Map<Object, Map<String, Object>> leftResult = leftTask.join();
            Map<Object, Map<String, Object>> rightResult = rightTask.join();
            result.putAll(rightResult);
            result.putAll(leftResult);
        }
        return result;
    }
}
